#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <pthread.h>
#include <time.h>

#include <sys/time.h>
#include <sys/select.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include "type.h"
#include "crc32.h"
#include "debug.h"
#include "dbsync.h"

#define DBSYNC_DBG_TAG  "dbsync"

#pragma pack(1)
struct header {
	uint8_t  cmd;
	uint8_t  sta;
	uint16_t len;
};

struct packet {
	struct header hdr;
	uint8_t       *buf;
	uint32_t      crc;
};
#pragma pack()

#define CMD_FILE_INFO	(0x00)
#define CMD_FILE_DATA	(0x01)

#define STA_OK		(0x00)
#define STA_CRC_ERR	(0x01)
#define STA_CMD_ERR	(0x02)

static int pre_role = UNKNOWN;
static int cur_role = UNKNOWN;
static int pre_maintain = 0;
static int cur_maintain = 0;

static pthread_t tid = 0;
static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

int dbsync_net_set_nonblock(int fd)
{
	int ret;
	int flags;

	if ((flags = fcntl(fd, F_GETFL)) == -1) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "set nonblock failed");
		return -1;
	}

	flags |= O_NONBLOCK;
	if ((ret = fcntl(fd, F_SETFL, flags)) == -1) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "set nonblock failed");
		return -1;
	}

	return 0;
}

int dbsync_net_set_block(int fd)
{
	int ret;
	int flags;

	if ((flags = fcntl(fd, F_GETFL)) == -1) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "set nonblock failed");
		return -1;
	}

	flags &= ~O_NONBLOCK;
	if ((ret = fcntl(fd, F_SETFL, flags)) == -1) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "set nonblock failed");
		return -1;
	}

	return 0;
}

int dbsync_server_init(const char *ip, uint16_t port)
{
	int ret;
	int sockfd;
	int optval;
	struct sockaddr_in srvaddr;

	sockfd = socket(AF_INET, SOCK_STREAM, 0);
	if (sockfd == -1) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
		return -1;
	}

	optval = 1;
	ret = setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
	if (ret == -1) {
		close(sockfd);
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
		return -1;
	}

	srvaddr.sin_family = AF_INET;
	ret = inet_aton(ip, &srvaddr.sin_addr);
	if (ret == 0) {
		close(sockfd);
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "invalid ip address");
		return -1;
	}
	srvaddr.sin_port = htons(SYNC_PORT);

	ret = bind(sockfd, (const struct sockaddr *)&srvaddr, sizeof(struct sockaddr_in));
	if (ret == -1) {
		close(sockfd);
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
		return -1;
	}

	ret = listen(sockfd, 0);
	if (ret == -1) {
		close(sockfd);
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
		return -1;
	}

	return sockfd;
}

int dbsync_server_new_conn(int sockfd)
{
	int connfd;
	struct sockaddr_in cliaddr;
	socklen_t addrlen;

	connfd = accept(sockfd, (struct sockaddr *)&cliaddr, &addrlen);
	if (connfd == -1) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
		return -1;
	}

	debug(INFO, DBSYNC_DBG_TAG, "a new client is comming: %s:%d\n", inet_ntoa(cliaddr.sin_addr), ntohs(cliaddr.sin_port));

	return connfd;
}

int dbsync_client_init(const char *ip, uint16_t port)
{
	int ret;
	int sockfd;
	fd_set set;
	struct sockaddr_in srvaddr;

	sockfd = socket(AF_INET, SOCK_STREAM, 0);
	if (sockfd == -1) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
		return -1;
	}

	srvaddr.sin_family = AF_INET;
	ret = inet_aton(ip, &srvaddr.sin_addr);
	if (ret == 0) {
		close(sockfd);
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "invalid ip address");
		return -1;
	}
	srvaddr.sin_port = htons(port);

	dbsync_net_set_nonblock(sockfd);
	ret = connect(sockfd, (const struct sockaddr *)&srvaddr, sizeof(struct sockaddr_in));
	if (ret == -1) {
		if (errno == EINPROGRESS) {
			struct timeval tv;

			FD_ZERO(&set);
			FD_SET(sockfd, &set);
			tv.tv_sec = 10;
			tv.tv_usec = 0;

			ret = select(sockfd + 1, NULL, &set, NULL, &tv);
			if (ret == -1) {
				debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
				return -1;
			} else if (ret == 0) {
				debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "connect to server timeout");
				return -1;
			} else {
				debug(INFO, DBSYNC_DBG_TAG, "%s\n", "connect to server success");
				return sockfd;
			}
		} else {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
			return -1;
		}
	}

	return sockfd;
}

int dbsync_net_read_a_seg(int connfd, uint8_t *buf, uint16_t len, int seconds)
{
	int ret;

	if (seconds) {
		fd_set set;
		struct timeval tv;
again:
		FD_ZERO(&set);
		FD_SET(connfd, &set);
		tv.tv_sec = seconds;
		tv.tv_usec = 0;
		dbsync_net_set_nonblock(connfd);

		if ((ret = select(connfd + 1, &set, NULL, NULL, &tv)) == -1) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
			return -1;
		}
		if (ret == 0) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "net read time out");
			return -1;
		}

		ret = read(connfd, buf, len);
		if (ret > 0) {
			return ret;
		} else if (ret == 0) {
			debug(INFO, DBSYNC_DBG_TAG, "%s\n", "closed by peer");
			return -1;
		} else if (ret == -1 && errno == EAGAIN) {
			goto again;
		} else {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
			return -1;
		}
	} else {
		dbsync_net_set_block(connfd);
		ret = read(connfd, buf, len);
		if (ret > 0) {
			return ret;
		} else if (ret == 0) {
			debug(INFO, DBSYNC_DBG_TAG, "%s\n", "closed by peer");
			return -1;
		} else {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
			return -1;
		}
	}
}

int dbsync_net_write_a_seg(int connfd, uint8_t *buf, uint16_t len, int seconds)
{
	int ret;

	if (seconds) {
		fd_set set;
		struct timeval tv;

		FD_ZERO(&set);
		FD_SET(connfd, &set);
		tv.tv_sec = seconds;
		tv.tv_usec = 0;
		dbsync_net_set_nonblock(connfd);

		if ((ret = select(connfd + 1, NULL, &set, NULL, &tv)) == -1) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
			return -1;
		}
		if (ret == 0) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "net write time out");
			return -1;
		}

		ret = write(connfd, buf, len);
		if (ret == -1) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
			return -1;
		}

		return ret;
	} else {
		dbsync_net_set_block(connfd);
		ret = write(connfd, buf, len);
		if (ret == -1) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
			return -1;
		}

		return ret;
	}
}

int dbsync_net_read(int connfd, uint8_t *buf, uint16_t len, int seconds)
{
	int ret;
	uint16_t total = 0;

	while (total != len) {
		ret = dbsync_net_read_a_seg(connfd, buf + total, len - total, seconds);
		if (ret == -1 || ret == 0)
			return total;
		total += ret;
	}

	return total;
}

int dbsync_net_write(int connfd, uint8_t *buf, uint16_t len, int seconds)
{
	int ret;
	uint16_t total = 0;

	while (total != len) {
		ret = dbsync_net_write_a_seg(connfd, buf + total, len - total, seconds);
		if (ret == -1)
			return total;
		total += ret;
	}

	return total;
}

uint32_t dbsync_net_gen_packet_crc(struct packet *p)
{
	uint32_t crc = 0xffffffff;

	crc = crc32((const unsigned char *)p, sizeof(struct header), crc);
	crc = crc32(p->buf, p->hdr.len, crc);

	return crc;
}

int dbsync_net_snd_a_packet(int connfd, struct packet *p, int seconds);

int dbsync_net_check_packet(struct packet *p, int fd)
{
	struct packet ack;
	uint32_t crc;

	if (p->hdr.cmd != CMD_FILE_INFO && p->hdr.cmd != CMD_FILE_DATA) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "unsupported command");
		ack.hdr.cmd = p->hdr.cmd;
		ack.hdr.sta = STA_CMD_ERR;
		ack.hdr.len = 0;
		ack.buf = NULL;
		dbsync_net_snd_a_packet(fd, &ack, 10);
		return -1;
	}

	crc = dbsync_net_gen_packet_crc(p);
	if (crc != p->crc) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "check packet crc failed");
		ack.hdr.cmd = p->hdr.cmd;
		ack.hdr.sta = STA_CRC_ERR;
		ack.hdr.len = 0;
		ack.buf = NULL;
		dbsync_net_snd_a_packet(fd, &ack, 10);
		return -1;
	}

	return 0;
}

int dbsync_net_rcv_a_packet(int connfd, struct packet *p, int seconds)
{
	int ret;
	unsigned int length;
	uint8_t *buf;

	p->buf = NULL;
	ret = dbsync_net_read(connfd, (uint8_t *)p, sizeof(struct header), seconds);
	if (ret != sizeof(struct header)) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "read packet header failed");
		goto error;
	}
	debug(DEBUG, DBSYNC_DBG_TAG, "%s\n", "packet header is:");
	debug(DEBUG, DBSYNC_DBG_TAG, "\tcmd: %#02x\n", p->hdr.cmd);
	debug(DEBUG, DBSYNC_DBG_TAG, "\tsta: %#02x\n", p->hdr.sta);
	debug(DEBUG, DBSYNC_DBG_TAG, "\tlen: %#04x\n", p->hdr.len);

	if (p->hdr.len) {
		p->buf = (uint8_t *)malloc(p->hdr.len);
		ret = dbsync_net_read(connfd, (uint8_t *)p->buf, p->hdr.len, seconds);
		if (ret != p->hdr.len) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "read packet data failed");
			goto error;
		}
		debug(DEBUG, DBSYNC_DBG_TAG, "%s\n", "read packet data successed");
	} else {
		p->buf = NULL;
	}

	ret = dbsync_net_read(connfd, (uint8_t *)&p->crc, sizeof(uint32_t), seconds);
	if (ret != sizeof(uint32_t)) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "read packet crc failed");
		goto error;
	}
	debug(DEBUG, DBSYNC_DBG_TAG, "\tcrc: %#04x\n", p->crc);
	
	return dbsync_net_check_packet(p, connfd);
error:
	return -1;
}

int dbsync_net_snd_a_packet(int connfd, struct packet *p, int seconds)
{
	int ret;

	p->crc = dbsync_net_gen_packet_crc(p);
	debug(DEBUG, DBSYNC_DBG_TAG, "%s\n", "packet header is:");
	debug(DEBUG, DBSYNC_DBG_TAG, "\tcmd: %#02x\n", p->hdr.cmd);
	debug(DEBUG, DBSYNC_DBG_TAG, "\tsta: %#02x\n", p->hdr.sta);
	debug(DEBUG, DBSYNC_DBG_TAG, "\tlen: %#04x\n", p->hdr.len);
	debug(DEBUG, DBSYNC_DBG_TAG, "\tcrc: %#04x\n", p->crc);
	ret = dbsync_net_write(connfd, (uint8_t *)p, sizeof(struct header), seconds);

	if (ret != sizeof(struct header)) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "write packet header failed");
		goto error;
	}

	if (p->hdr.len) {
		ret = dbsync_net_write(connfd, (uint8_t *)p->buf, p->hdr.len, seconds);
		if (ret != p->hdr.len) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "write packet data failed");
			goto error;
		}
		debug(DEBUG, DBSYNC_DBG_TAG, "%s\n", "write packet data successed");
	}

	ret = dbsync_net_write(connfd, (uint8_t *)&p->crc, sizeof(uint32_t), seconds);
	if (ret != sizeof(uint32_t)) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "write packet crc failed");
		goto error;
	}
	
	return 0;
error:
	return -1;
}

int dbsync_copy_file(const char *from, const char *to)
{
	int ret;
	int fd1, fd2;
	char buf[1024];

	debug(DEBUG, DBSYNC_DBG_TAG, "%s\n", "copy file started");

	fd1 = open(from, O_RDONLY);
	fd2 = open(to, O_WRONLY | O_TRUNC);
	if (fd1 == -1 || fd2 == -1) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "copy file failed");
		return -1;
	}

	do {
		ret = read(fd1, buf, sizeof(buf));
		if (ret < 0) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "read from file failed");
			return ret;
		} else if (ret == 0) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "read file finished");
			break;
		}
		ret = write(fd2, buf, ret);
		if (ret <= 0) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "write to file failed");
			return ret;
		}
	} while (ret);

	debug(DEBUG, DBSYNC_DBG_TAG, "%s\n", "copy file finished");
	return 0;
}

int dbsync_server_sync_file(int connfd) {
	int fd;
	int ret;
	uint32_t length;
	uint32_t total;
	struct packet req, ack;

	while (1) {
		ret = dbsync_net_rcv_a_packet(connfd, &req, 0);
		if (ret == -1) {
			if (req.buf)
				free(req.buf);
			return -1;
		}

		/* get file length */
		if (req.hdr.cmd != CMD_FILE_INFO) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "not a valid cmd for this time");
			if (req.buf)
				free(req.buf);
			ack.hdr.cmd = req.hdr.cmd;
			ack.hdr.sta = STA_CMD_ERR;
			ack.hdr.len = 0;
			ack.buf = NULL;
			ret = dbsync_net_snd_a_packet(connfd, &ack, 10);
			if (ret == -1)
				return -1;
			continue;
		} else {
			debug(INFO, DBSYNC_DBG_TAG, "file length: %d\n", *(uint32_t *)req.buf);
			length = *(uint32_t *)req.buf;
			free(req.buf);
			ack.hdr.cmd = req.hdr.cmd;
			ack.hdr.sta = STA_OK;
			ack.hdr.len = 0;
			ack.buf = NULL;
			ret = dbsync_net_snd_a_packet(connfd, &ack, 10);
			if (ret == -1)
				return -1;
		}

		fd = open("/tmp/data.db", O_CREAT | O_WRONLY | O_TRUNC, 0664);
		if (fd == -1) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
			return -1;
		}

		total = 0;
		while (total < length) {
			/* get file data */
			ret = dbsync_net_rcv_a_packet(connfd, &req, 10);
			if (ret == -1) {
				if (req.buf)
					free(req.buf);
				return -1;
			}

			if (req.hdr.cmd != CMD_FILE_DATA) {
				debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "not a valid cmd for this time");
				if (req.buf)
					free(req.buf);
				ack.hdr.cmd = req.hdr.cmd;
				ack.hdr.sta = STA_CMD_ERR;
				ack.hdr.len = 0;
				ack.buf = NULL;
				ret = dbsync_net_snd_a_packet(connfd, &ack, 10);
				if (ret == -1)
					return -1;
			} else {
				ret = write(fd, req.buf, req.hdr.len);
				if (ret == -1) {
					debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
					free(req.buf);
					close(fd);
					return -1;
				}
				free(req.buf);

				ack.hdr.cmd = req.hdr.cmd;
				ack.hdr.sta = STA_OK;
				ack.hdr.len = 0;
				ack.buf = NULL;
				ret = dbsync_net_snd_a_packet(connfd, &ack, 10);
				if (ret == -1)
					return -1;

				total += req.hdr.len;
			}
		}

		close(fd);

		pthread_mutex_lock(&lock);
		ret = dbsync_copy_file("/tmp/data.db", DB_FILE);
		pthread_mutex_unlock(&lock);

		return ret;
	}
}

void dbsync_file_server_clean(void *arg)
{
	int fd = *(int *)arg;
	close(fd);
}

void *dbsync_file_server(void *arg)
{
	int ret;
	int res;
	int sockfd;
	int connfd;

	while (1) {
		sockfd = dbsync_server_init("0.0.0.0", SYNC_PORT);
		if (sockfd == -1) {
			res = 1;
			pthread_exit(&res);
		}
		pthread_cleanup_push(dbsync_file_server_clean, &sockfd);

		connfd = dbsync_server_new_conn(sockfd);
		/* only one client */
		close(sockfd);
		pthread_cleanup_pop(0);

		if (connfd == -1) {
			pthread_mutex_lock(&lock);
			if (cur_role == STANDBY) {
				pthread_mutex_unlock(&lock);
				continue;
			} else {
				pthread_mutex_unlock(&lock);
				res = 2;
				break;
			}
		}

		pthread_cleanup_push(dbsync_file_server_clean, &connfd);
		ret = dbsync_server_sync_file(connfd);
		close(connfd);
		pthread_cleanup_pop(0);

		if (ret == -1) {
			pthread_mutex_lock(&lock);
			if (cur_role == STANDBY) {
				pthread_mutex_unlock(&lock);
				continue;
			} else {
				pthread_mutex_unlock(&lock);
				res = 3;
				break;
			}
		}

	}

	pthread_mutex_destroy(&lock);
	pthread_exit(&res);
}

int dbsync_start_server()
{
	int ret;

	if (tid != 0) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "sync server is already running");
		return 0;
	}

	pthread_mutex_init(&lock, NULL);
	ret = pthread_create(&tid, NULL, &dbsync_file_server, NULL);
	if (ret != 0) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(ret));
		return -1;
	} else {
		debug(INFO, DBSYNC_DBG_TAG, "%s\n", "sync server started successfully");
		return 0;
	}
}

int dbsync_stop_server()
{
	int ret;
	int sockfd;
	struct packet req, ack;
	void *res;

	if (tid == 0) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "sync server has stopped running");
		return 0;
	}

	ret = pthread_cancel(tid);
	if (ret) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", strerror(ret));
	}

	ret = pthread_join(tid, &res);
	pthread_mutex_destroy(&lock);
	if (ret != 0) {
		errno = ret;
		debug(INFO, DBSYNC_DBG_TAG, "%s\n", strerror(errno));
		return -1;
	}

	tid = 0;
	debug(INFO, DBSYNC_DBG_TAG, "%s\n", "stop sync server success");
	return 0;
}

int dbsync_do_sync(const char *ip, const char *filename)
{
	int ret;
	int sockfd;
	int filefd;
	char buf[1024];
	struct stat st;
	struct packet req, ack;

	if ((sockfd = dbsync_client_init(ip, SYNC_PORT)) == -1)
		return -1;

	if ((filefd = open(filename, O_RDONLY)) == -1) {
		close(sockfd);
		return -1;
	}

	ret = fstat(filefd, &st);
	if (ret == -1) {
		close(sockfd);
		close(filefd);
		return -1;
	}

	*(uint32_t *)buf = st.st_size;

	req.hdr.cmd = CMD_FILE_INFO;
	req.hdr.sta = 0;
	req.hdr.len = sizeof(uint32_t);
	req.buf = buf;
	ret = dbsync_net_snd_a_packet(sockfd, &req, 10);
	if (ret == -1) {
		close(sockfd);
		close(filefd);
		return -1;
	}

	ack.buf = NULL;
	ret = dbsync_net_rcv_a_packet(sockfd, &ack, 10);
	if (ret == -1) {
		if (ack.buf)
			free(ack.buf);
		close(sockfd);
		close(filefd);
		return -1;
	}

	if (ack.hdr.cmd != CMD_FILE_INFO || ack.hdr.sta != STA_OK) {
		debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "not a valid ack for this time");
		if (ack.buf)
			free(ack.buf);
		close(sockfd);
		close(filefd);
		return -1;
	}

	while (1) {
		ret = read(filefd, buf, sizeof(buf));
		if (ret == -1) {
			debug(ERROR, DBSYNC_DBG_TAG, "%s\n", "read file failed");
			close(sockfd);
			close(filefd);
			return -1;
		} else if (ret == 0) {
			debug(INFO, DBSYNC_DBG_TAG, "%s\n", "read file finished");
			close(sockfd);
			close(filefd);
			return 0;
		} else {
			req.hdr.cmd = CMD_FILE_DATA;
			req.hdr.sta = 0;
			req.hdr.len = ret;
			req.buf = buf;
			ret = dbsync_net_snd_a_packet(sockfd, &req, 10);
			if (ret == -1) {
				close(sockfd);
				close(filefd);
				return -1;
			}
			ack.buf = NULL;
			ret = dbsync_net_rcv_a_packet(sockfd, &ack, 10);
			if (ret == -1) {
				if (ack.buf)
					free(ack.buf);
				close(sockfd);
				close(filefd);
				return -1;
			}
			if (ack.hdr.cmd != CMD_FILE_DATA || ack.hdr.sta != STA_OK) {
				if (ack.buf)
					free(ack.buf);
				close(sockfd);
				close(filefd);
				return -1;
			}
		}
	}
}

